package org.atguigu.gmall.realtime.app.dwd.db;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.atguigu.gmall.realtime.app.BaseSQLApp;
import org.atguigu.gmall.realtime.common.Constant;
import org.atguigu.gmall.realtime.util.SQLUtil;

/**
 * Project:gmallRealTime
 * Package:org.atguigu.gmall.realtime.app.dwd.db
 * Author: ZengHaiFeng
 * CreateTime:2023/4/25 19:59:02
 * Description: Todo
 * Version：1.0
 */
public class Dwd_05_DwdTradeOrderCancelDetail extends BaseSQLApp {
    public static void main(String[] args) {
        new Dwd_05_DwdTradeOrderCancelDetail().init(30005,2,"Dwd_05_DwdTradeOrderCancelDetail");
    }
    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tbe) {
        // read data from dwd_trade_order_detail
        tbe.executeSql("create table `dwd_trade_order_detail`(" +
                " id string, " +
                " order_id string, " +
                " user_id string, " +
                " sku_id string, " +
                " sku_name string, " +
                " province_id string, " +
                " activity_id string, " +
                " activity_rule_id string, " +
                " coupon_id string, " +
                " date_id string, " +
                " create_time string, " +
                " sku_num string, " +
                " split_original_amount string, " +
                " split_activity_amount string, " +
                " split_coupon_amount string, " +
                " split_total_amount string, " +
                " ts bigint ) " +
                SQLUtil.getKafkaSourceDDL(Constant.TOPIC_DWD_TRADE_ORDER_DETAIL,
                "Dwd_05_DwdTradeOrderCancelDetail"));

        // read data from ods_db
        readOdsDb(tbe,"Dwd_05_ods_db");

        // filter order cancel data from ods_ds and order_info
        Table orderCancel = tbe.sqlQuery("select" +
                " `data`['id'] id, " +
                " `data`['operate_time'] order_cancel_time, " +
                " `ts` " +
                " from ods_db where " +
                " `database`='gmall2023' and " +
                " `table`='order_info' and " +
                " `type`='update' and " +
                " `old`['order_status']='1001' and " +
                " `data`['order_status']='1003'");
        tbe.createTemporaryView("order_cancel",orderCancel);

        // join
        Table result = tbe.sqlQuery("select " +
                " od.id, " +
                " od.order_id, " +
                " od.user_id, " +
                " od.sku_id, " +
                " od.sku_name, " +
                " od.province_id, " +
                " od.activity_id, " +
                " od.activity_rule_id, " +
                " od.coupon_id, " +
                " date_format(oc.order_cancel_time,'yyyy-MM-dd') as order_cancel_time_id, " +
                " oc.order_cancel_time, " +
                " od.sku_num, " +
                " od.split_original_amount, " +
                " od.split_activity_amount, " +
                " od.split_coupon_amount, " +
                " od.split_total_amount, " +
                " oc.ts " +
                " from order_cancel as oc join dwd_trade_order_detail as od " +
                " on oc.id=od.order_id");

        // output the final result into kafka topic
        tbe.executeSql("create table dwd_trade_order_cancel_detail(" +
                " id string, " +
                " order_id string, " +
                " user_id string, " +
                " sku_id string, " +
                " sku_name string, " +
                " province_id string, " +
                " activity_id string, " +
                " activity_rule_id string, " +
                " coupon_id string, " +
                " order_cancel_time_id string, " +
                " order_cancel_time string, " +
                " sku_num string, " +
                " split_original_amount string, " +
                " split_activity_amount string, " +
                " split_coupon_amount string, " +
                " split_total_amount string, " +
                " ts bigint" +
                ")" + SQLUtil.getKafkaSinkDDL(Constant.TOPIC_DWD_TRADE_ORDER_CANCEL_DETAIL));
        result.executeInsert("dwd_trade_order_cancel_detail");

    }
}
